package cn.doitedu.profile.graph

import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-06-21
 * @desc 图计算demo
 */
object Demo {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("demo")
      .master("local")
      .getOrCreate()


    val csv: DataFrame = spark.read.option("header",true).csv("profile/data")


    // spark  graphx 图计算算法的调用流程

    // 先提取出数据中的所有 “点” ，放在一个rdd中
    import org.apache.spark.sql.functions._
    import spark.implicits._
    val idsDataframe = csv.selectExpr("array(phone,name,email) as arr").select(explode($"arr") as "id")

    // idsDataframe.show(100,false)
    val vertices: RDD[(Long, String)] = idsDataframe.rdd.map(row => {
    val indentity = row.getAs[String]("id")
      //(点的全局id[Long]，点的属性数据)
      (indentity.hashCode.toLong, indentity)
    })


    vertices.foreach(println)
    println("---------------------------------")

    // 提取出数据中的所有“边”，放在一个rdd中
    val edges: RDD[Edge[String]] = csv.rdd.flatMap(row=>{
      val phone = row.getAs[String]("phone")
      val name = row.getAs[String]("name")
      val email = row.getAs[String]("email")
      val indetities = Array(name, email)

      val edgeArr: Array[Edge[String]] = for (elem <- indetities)  yield {
        Edge(phone.hashCode.toLong,elem.hashCode.toLong,"")
      }

      edgeArr

    })

    // 然后将点rdd和边rdd，构造成一个图graph
    val graph = Graph(vertices, edges)

    // 然后就可以在图对象上调用各种写好的算法
    val resultGraph = graph.connectedComponents()

    val resultVertices = resultGraph.vertices
    resultVertices.foreach(println)


    // 解析算法返回的结果


    // 最终要的结果：
    // 1,900  (代表行哥的总金额）
    // 2,1100 （代表娜姐的总金额）


    spark.close()
  }

}
